package org.example

import org.apache.spark.sql.SparkSession

//
//import org.apache.spark.sql.SparkSession
//
//import java.lang.reflect.Array
//
object s3 {
  def main(args: Array[String]): Unit = {
    // 创建spark运行环境
   /* val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
    */

    val spark = SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
    val sc = spark.sparkContext
//// 1.创建RDD 不指定分区就会打散
//    val rdd1 =sc.makeRDD(List(1,5,3,7,6))
//    val rdd2 = sc.parallelize(List(('a',90),('b',78),('c',66),('b',78)))
//// 2.spark数据分析方法
    val rdd3 = sc.makeRDD(List(1,2,3,4,5),1)
////交集intersection
////    rdd1.intersection(rdd3).foreach(println)
////并集union
////    rdd1.union(rdd3 .foreach(println)
////差集subtract 站在rdd1的角度上看是 6 7
////    rdd1.subtract(rdd3).foreach(println)
////   zip拉链 分区数要一致 形成元组（key,value)
////   rdd1.zip(rdd3).foreach(println)
////  cartesian 笛卡尔积 1跟3中的每个元素都会重新组合成一组，5*5=25组
//     rdd1.cartesian(rdd3).foreach(println)
////   fiter rdd2中筛选出大于等于90的数据
//    rdd2.filter(_._2 >= 90).foreach(println)
////去重distinct
//    rdd2.distinct().foreach(println)
//// 3.输出结果
//// rdd1.foreach(println)
//// rdd2.collect( .foreach(println)
//// 键值对(key value)
//   reduceByKey  聚合 分组
    val data1 = sc.makeRDD(List(('a',90),('b',78),('c',66),('b',78)),2)
    val data2 = sc.makeRDD(List(('a',5),('b',3),('c',4),('a',6)),2)
//    data1.reduceByKey((x,y) => x + y).foreach(println)

//  groupByKey 根据key分组
//    data1.groupByKey().foreach(println)
//      data1.groupBy(_._1).foreach(println)

//   join 相同的 key 和 value 组成新的元组 如果岑在多个相同的key 会依次匹配
//      data1.join(data2).foreach(println)
//   以左边为主，右边可能匹配不到
        data1.leftOuterJoin(data2).foreach(println)
//   combineByKey  联合
        data1.combineByKey(
          v => (v,1),
          (t:(Int,Int),v) => {
            (t._1+v,t._2+1)
          },
          (t1: (Int, Int),t2:(Int,Int)) => {
            (t1._1+t2._1,t1._2+t2._2)
          }
        ).foreach(println)
//   集合
    data1.aggregateByKey(0)(math.max(_,_),_+_).foreach(println)
//    结束后释放资源
      sc.stop()

    }
//
  }
